// 默认数据转换脚本
// 用于将Kafka消息转换为Hive表格式

import groovy.json.JsonSlurper
import groovy.json.JsonBuilder
import java.text.SimpleDateFormat

/**
 * 数据转换函数
 * @param data 原始数据（JSON字符串）
 * @return 转换后的数据（JSON字符串）
 */
def transform(data) {
    try {
        // 解析JSON数据
        def jsonSlurper = new JsonSlurper()
        def jsonData = jsonSlurper.parseText(data)
        
        // 添加处理时间戳
        def now = new Date()
        def dateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss.SSS')
        jsonData.processTime = dateFormat.format(now)
        jsonData.processTimestamp = now.getTime()
        
        // 添加数据来源标识
        jsonData.source = 'kafka'
        jsonData.gateway = 'data-gateway'
        
        // 数据质量检查
        if (!jsonData.containsKey('id') || jsonData.id == null) {
            jsonData.id = UUID.randomUUID().toString()
        }
        
        // 添加数据大小信息
        jsonData.originalDataSize = data.getBytes().length
        
        // 数据标准化处理
        standardizeData(jsonData)
        
        // 转换为JSON字符串
        def jsonBuilder = new JsonBuilder(jsonData)
        return jsonBuilder.toString()
        
    } catch (Exception e) {
        // 如果解析失败，返回包含错误信息的原始数据
        def errorData = [
            originalData: data,
            processTime: new SimpleDateFormat('yyyy-MM-dd HH:mm:ss.SSS').format(new Date()),
            source: 'kafka',
            gateway: 'data-gateway',
            error: true,
            errorMessage: e.getMessage(),
            id: UUID.randomUUID().toString()
        ]
        
        def jsonBuilder = new JsonBuilder(errorData)
        return jsonBuilder.toString()
    }
}

/**
 * 数据标准化处理
 * @param jsonData 解析后的JSON数据
 */
def standardizeData(jsonData) {
    // 标准化时间字段
    if (jsonData.containsKey('timestamp')) {
        try {
            def timestamp = jsonData.timestamp
            if (timestamp instanceof String) {
                // 尝试解析时间字符串
                def dateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss')
                def date = dateFormat.parse(timestamp)
                jsonData.dataTimestamp = dateFormat.format(date)
            } else if (timestamp instanceof Number) {
                // 时间戳转换为日期字符串
                def date = new Date(timestamp.longValue())
                def dateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss')
                jsonData.dataTimestamp = dateFormat.format(date)
            }
        } catch (Exception e) {
            // 时间解析失败，使用当前时间
            def dateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss')
            jsonData.dataTimestamp = dateFormat.format(new Date())
        }
    } else {
        // 没有时间字段，使用当前时间
        def dateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss')
        jsonData.dataTimestamp = dateFormat.format(new Date())
    }
    
    // 标准化数值字段
    standardizeNumericFields(jsonData)
    
    // 标准化字符串字段
    standardizeStringFields(jsonData)
}

/**
 * 标准化数值字段
 * @param jsonData 解析后的JSON数据
 */
def standardizeNumericFields(jsonData) {
    // 处理常见的数值字段
    def numericFields = ['amount', 'price', 'quantity', 'count', 'value', 'score', 'rate']
    
    numericFields.each { field ->
        if (jsonData.containsKey(field)) {
            try {
                def value = jsonData[field]
                if (value instanceof String) {
                    jsonData[field] = Double.parseDouble(value)
                }
            } catch (Exception e) {
                // 数值转换失败，保持原值
            }
        }
    }
}

/**
 * 标准化字符串字段
 * @param jsonData 解析后的JSON数据
 */
def standardizeStringFields(jsonData) {
    // 处理常见的字符串字段
    def stringFields = ['name', 'description', 'title', 'content', 'message']
    
    stringFields.each { field ->
        if (jsonData.containsKey(field)) {
            def value = jsonData[field]
            if (value != null) {
                // 去除首尾空格
                jsonData[field] = value.toString().trim()
            }
        }
    }
}

// 返回转换函数
return this
